package com.lagou.rpc.consumer.proxy;

import com.alibaba.fastjson.JSON;
import com.lagou.rpc.api.IUserService;
import com.lagou.rpc.common.RpcRequest;
import com.lagou.rpc.common.RpcResponse;
import com.lagou.rpc.consumer.client.RpcClient;
import com.lagou.rpc.zookeeper.ZookeeperClient;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.springframework.util.CollectionUtils;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 客户端代理类-创建代理对象
 * 1.封装request请求对象
 * 2.创建RpcClient对象
 * 3.发送消息
 * 4.返回结果
 */
public class RpcClientProxy {

    private static Map<String, RpcClient> ipAddressRpcClientMap = new HashMap<>();

    private static ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);

    private static long costTime = 0l;

    private static ZookeeperClient zookeeperClient = new ZookeeperClient();


    private static RpcClient currentRpcClient;

    private static Map<String, String> serverLastCallTimeMap = new HashMap<>();

    private static final String WATCH_PATH = "/" + IUserService.class.getName();

    static {
        zookeeperClient.init();
        List<String> childrenList = zookeeperClient.getChildrenKeys(WATCH_PATH);
        //2.创建RpcClient对象
        if (!CollectionUtils.isEmpty(childrenList)) {
            for (String ipAndPort : childrenList) {
                RpcClient rpcClient = createRpcClient(ipAndPort);
                ipAddressRpcClientMap.put(ipAndPort, rpcClient);
            }
        }

        PathChildrenCacheListener cacheListener = (client, event) -> {
            String fullPath;
            String ipAndPort;
            switch (event.getType()) {
                case CHILD_ADDED:
                    fullPath = event.getData().getPath();
                    ipAndPort = fullPath.substring(fullPath.lastIndexOf("/") + 1);
                    RpcClient rpcClient = createRpcClient(ipAndPort);
                    ipAddressRpcClientMap.put(ipAndPort, rpcClient);
                    break;
                case CHILD_REMOVED:
                    fullPath = event.getData().getPath();
                    ipAndPort = fullPath.substring(fullPath.lastIndexOf("/") + 1);
                    ipAddressRpcClientMap.remove(ipAndPort);
                    break;
                default:
                    return;
            }
        };
        zookeeperClient.watch(WATCH_PATH, cacheListener);

        executorService.scheduleWithFixedDelay(() -> {

            for (Map.Entry<String, String> entry : serverLastCallTimeMap.entrySet()) {
                long lastCallTime = Long.parseLong(entry.getValue().split(":")[0]);
                if (System.currentTimeMillis() - lastCallTime > 5000l) {
                    serverLastCallTimeMap.remove(entry.getKey());
                }
            }

        }, 5000, 5000, TimeUnit.MILLISECONDS);
    }

    private static RpcClient createRpcClient(String ipAndPort) {
        String ip = ipAndPort.split(":")[0];
        int port = Integer.parseInt(ipAndPort.split(":")[1]);
        return new RpcClient(ip, port);
    }

    public static Object createProxy(Class serviceClass) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                new Class[]{serviceClass}, new InvocationHandler() {
                    @Override
                    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                        //1.封装request请求对象
                        RpcRequest rpcRequest = new RpcRequest();
                        rpcRequest.setRequestId(UUID.randomUUID().toString());
                        rpcRequest.setClassName(method.getDeclaringClass().getName());
                        rpcRequest.setMethodName(method.getName());
                        rpcRequest.setParameterTypes(method.getParameterTypes());
                        rpcRequest.setParameters(args);
                        if (ipAddressRpcClientMap.isEmpty()) {
                            throw new RuntimeException("无服务");
                        }
                        RpcClient rpcClient = null;
                        long minCostTime = Long.MIN_VALUE;
                        for (Map.Entry<String, String> entry : serverLastCallTimeMap.entrySet()) {
                            long time = Long.parseLong(entry.getValue().split(":")[1]);
                            if (time < minCostTime){
                                rpcClient = ipAddressRpcClientMap.get(entry.getKey());
                            }
                        }
                        if (rpcClient == null){
                            rpcClient = ipAddressRpcClientMap.values().iterator().next();
                        }
                        currentRpcClient = rpcClient;

                        try {
                            //3.发送消息
                            long startTime = System.currentTimeMillis();
                            Object responseMsg = rpcClient.send(JSON.toJSONString(rpcRequest));
                            RpcResponse rpcResponse = JSON.parseObject(responseMsg.toString(), RpcResponse.class);
                            if (rpcResponse.getError() != null) {
                                throw new RuntimeException(rpcResponse.getError());
                            }
                            long endTime = System.currentTimeMillis();
                            costTime = endTime - startTime;
                            serverLastCallTimeMap.put(currentRpcClient.getIp() + ":" + currentRpcClient.getPort(), +startTime + ":" + costTime);
                            //4.返回结果
                            Object result = rpcResponse.getResult();
                            return JSON.parseObject(result.toString(), method.getReturnType());
                        } catch (Exception e) {
                            throw e;
                        } finally {
                            rpcClient.close();
                        }

                    }
                });
    }
}
